from celery import shared_task
from django.conf import settings
from .models import OPCUASubscription, MQTTSubscription
from alarms.models import AlarmRecord
import asyncio
import paho.mqtt.client as mqtt
from asyncua import Client as OPCUAClient
import json
import logging
import threading
from django.db import transaction

logger = logging.getLogger(__name__)
"""
概述
    该文件定义了两个 Celery 异步任务，用于同步和管理 OPC UA 和 MQTT 订阅。这些任务通过 Celery 的 @shared_task 装饰
器定义，并使用 Django ORM 来操作数据库中的订阅记录。每个任务负责与相应的协议服务器（OPC UA 或 MQTT）建立连接，监听指定的
节点或主题，并根据配置条件触发报警。
"""


@shared_task
def sync_opcua_subscription(subscription_id, node_id=None, server_url=None, action='sync'):
    """
    同步 OPC UA 订阅任务
    
    参数:
        subscription_id: 订阅ID
        node_id: 节点ID（仅在同步时需要）
        server_url: 服务器URL（仅在同步时需要）
        action: 动作类型（'sync' 或 'cancel'）
    """
    try:
        if action == 'cancel':
            logger.info(f"取消 OPC UA 订阅 - ID: {subscription_id}")
            return

        subscription = OPCUASubscription.objects.get(id=subscription_id)
        if not subscription.is_active:
            logger.info(f"OPC UA 订阅未启用 - ID: {subscription_id}")
            return

        async def monitor_node():
            client = OPCUAClient(server_url)
            try:
                await client.connect()
                node = client.get_node(node_id)

                async def value_change_callback(node, val, data):  # 数据变化回调函k，当节点值发生变化时调用
                    value = float(val)
                    triggered = False

                    if subscription.condition == 'gt' and value > subscription.threshold_value:
                        triggered = True
                    elif subscription.condition == 'lt' and value < subscription.threshold_value:
                        triggered = True
                    elif subscription.condition == 'eq' and value == subscription.threshold_value:
                        triggered = True
                    elif subscription.condition == 'range':
                        if not (subscription.threshold_value <= value <= subscription.threshold_max):
                            triggered = True

                    if triggered:
                        message = subscription.alert_message.format(value=value)
                        AlarmRecord.objects.create(
                            subscription_type='opcua',
                            opcua_subscription=subscription,
                            value=value,
                            message=message,
                            name=subscription.name
                        )
                        logger.info(f"OPC UA 报警已触发 - ID: {subscription_id}, 值: {value}")

                # 订阅节点的数据变化事件，使用上述定义的回调函数处理数据变化
                handler = await node.subscribe_data_change(value_change_callback)
                # 循环检查订阅状态，直到订阅不再活跃
                while subscription.is_active:  # 定期检查订阅状态
                    await asyncio.sleep(5)
                # 当订阅不再活跃时，取消对节点数据变化的订阅
                await node.unsubscribe(handler)

            except Exception as e:
                logger.error(f"OPC UA 监控失败 - ID: {subscription_id}, 错误: {str(e)}", exc_info=True)
            finally:
                await client.disconnect()

        # 运行异步任务
        asyncio.run(monitor_node())

    except Exception as e:
        logger.error(f"OPC UA 订阅任务失败 - ID: {subscription_id}, 错误: {str(e)}", exc_info=True)


@shared_task
def sync_mqtt_subscription(subscription_id, topic=None, broker_url=None, action='sync'):
    """
    同步 MQTT 订阅任务
    
    参数:
        subscription_id: 订阅ID
        topic: MQTT主题（仅在同步时需要）
        broker_url: Broker URL（仅在同步时需要）
        action: 动作类型（'sync' 或 'cancel'）
    """
    try:
        if action == 'cancel':
            logger.info(f"取消 MQTT 订阅 - ID: {subscription_id}")
            return

        logger.info(f"开始同步 MQTT 订阅 - ID: {subscription_id}, Topic: {topic}, Broker: {broker_url}")

        subscription = MQTTSubscription.objects.get(id=subscription_id)
        if not subscription.is_active:
            logger.info(f"MQTT 订阅未启用 - ID: {subscription_id}")
            return

        def on_connect(client, userdata, flags, rc):
            """MQTT 连接回调"""
            if rc == 0:
                logger.info(f"MQTT 已成功连接 - ID: {subscription_id}, Topic: {topic}")
                client.subscribe(topic)
                logger.info(f"已订阅主题 - Topic: {topic}")
                connected.set()  # 设置连接成功标志
            else:
                logger.error(f"MQTT 连接失败 - ID: {subscription_id}, 错误代码: {rc}")
                connected.clear()  # 清除连接标志

        def on_disconnect(client, userdata, rc):
            """MQTT 断开连接回调"""
            connected.clear()  # 清除连接标志
            if rc != 0:
                logger.warning(f"MQTT 意外断开连接 - ID: {subscription_id}, 错误代码: {rc}")
                # 尝试重新连接
                try:
                    logger.info(f"尝试重新连接MQTT - ID: {subscription_id}")
                    client.reconnect()
                except Exception as e:
                    logger.error(f"MQTT 重连失败 - ID: {subscription_id}, 错误: {str(e)}")
            else:
                logger.info(f"MQTT 已断开连接 - ID: {subscription_id}")

        def on_subscribe(client, userdata, mid, granted_qos):
            """MQTT 订阅确认回调"""
            logger.info(f"MQTT 订阅已确认 - ID: {subscription_id}, Topic: {topic}, QoS: {granted_qos}")

        def on_message(client, userdata, msg):
            """MQTT 消息回调"""
            try:
                # logger.info(f"收到原始MQTT消息 - Topic: {msg.topic}, Payload: {msg.payload}")

                # 先解码消息内容
                try:
                    payload_str = msg.payload.decode('utf-8')
                except UnicodeDecodeError:
                    payload_str = msg.payload.decode('gbk')

                # logger.info(f"解码后的消息内容: {payload_str}")
                payload = json.loads(payload_str)
                logger.info(f"解析后的MQTT消息: {payload}")

                # 获取订阅对象
                subscription = MQTTSubscription.objects.get(id=subscription_id)
                if not subscription.is_active:
                    logger.info(f"MQTT订阅已禁用 - ID: {subscription_id}")
                    return

                # 使用事务确保数据库操作的原子性
                with transaction.atomic():
                    # 获取所有被启用的监控项
                    subscription_items = subscription.mqttsubscriptionitem_set.filter(is_active=True)
                    logger.info(f"获取到监控项数量: {subscription_items.count()}")

                    # 遍历每个监控项进行报警判断
                    for item in subscription_items:
                        try:
                            # 在values列表中查找匹配的数据项
                            value_data = None
                            values = payload.get('values', [])
                            logger.info(f"处理监控项 - ID: {item.id}, Values ID: {item.values_id}, 数据: {values}")

                            for data in values:
                                if data.get('id') == item.values_id:
                                    value_data = data
                                    logger.info(f"找到匹配的数据项 - Values ID: {item.values_id}, 数据: {data}")
                                    break

                            if value_data is None:
                                logger.info(f"未找到匹配的数据项 - 监控项ID: {item.id}, Values ID: {item.values_id}")
                                continue

                            # 获取数值并转换为float
                            value = float(value_data.get('v', 0))
                            logger.info(
                                f"获取到数值 - 监控项ID: {item.id}, 值: {value}, 条件: {item.condition}, 阈值: {item.threshold_value}")

                            # 通过 WebSocket 发送数据更新
                            from channels.layers import get_channel_layer
                            from asgiref.sync import async_to_sync
                            channel_layer = get_channel_layer()
                            # 构建消息内容
                            message_data = {
                                'type': 'subscription_data',
                                'data': {
                                    'type': 'data_update',
                                    'subscription_type': 'mqtt',
                                    'subscription_id': subscription.id,
                                    'item_id': item.id,
                                    'value': value,
                                    'quality': value_data.get('q'),
                                    'timestamp': value_data.get('t')
                                }
                            }
                            async_to_sync(channel_layer.group_send)(
                                f'user_{subscription.created_by.id}_subscriptions',
                                message_data
                            )
                            logger.info(f"已发送数据更新 - 监控项ID: {item.id}, 值: {value}")

                            # 报警条件判断
                            triggered = False
                            if item.condition == 'gt' and value > item.threshold_value:
                                triggered = True
                                logger.info(f"触发大于条件 - 值: {value} > {item.threshold_value}")
                            elif item.condition == 'lt' and value < item.threshold_value:
                                triggered = True
                                logger.info(f"触发小于条件 - 值: {value} < {item.threshold_value}")
                            elif item.condition == 'eq' and value == item.threshold_value:
                                triggered = True
                                logger.info(f"触发等于条件 - 值: {value} = {item.threshold_value}")
                            elif item.condition == 'range':
                                if item.threshold_value <= value <= item.threshold_max:
                                    triggered = True
                                    logger.info(
                                        f"触发范围条件 - 值: {value} 不在范围 [{item.threshold_value}, {item.threshold_max}] 内")

                            logger.info(f"报警条件判断结果 - 监控项ID: {item.id}, 触发状态: {triggered}")

                            if triggered:
                                # 创建报警记录
                                message = item.alert_message.format(value=value)
                                alarm = AlarmRecord.objects.create(
                                    subscription_type='mqtt',
                                    mqtt_subscription=subscription,
                                    mqtt_subscription_item=item,
                                    value=value,
                                    message=message,
                                    name=item.name
                                )
                                logger.info(
                                    f"MQTT 报警已触发 - 报警ID: {alarm.id}, 订阅ID: {subscription_id}, 监控项ID: {item.id}, 值: {value}")

                                # 构建消息内容
                                message_data = {
                                    'type': 'subscription_data',
                                    'data': {
                                        'type': 'alarm_update',
                                        'subscription_type': 'mqtt',
                                        'subscription_id': subscription.id,
                                        'item_id': item.id,
                                        'alarm_id': alarm.id,
                                        'message': message,
                                        'value': value,
                                        'timestamp': value_data.get('t')
                                    }
                                }
                                # 通过 WebSocket 发送报警更新
                                async_to_sync(channel_layer.group_send)(
                                    f"user_{subscription.created_by.id}_subscriptions",
                                    message_data
                                )
                                logger.info(f"已发送报警更新 - 监控项ID: {item.id}, 报警ID: {alarm.id}")

                        except Exception as e:
                            logger.error(
                                f"处理监控项时出错 - 订阅ID: {subscription_id}, 监控项ID: {item.id}, 错误: {str(e)}",
                                exc_info=True)

            except json.JSONDecodeError as e:
                logger.error(f"MQTT 消息JSON解析失败 - ID: {subscription_id}, 错误: {str(e)}", exc_info=True)
            except Exception as e:
                logger.error(f"MQTT 消息处理失败 - ID: {subscription_id}, 错误: {str(e)}", exc_info=True)

        # 创建MQTT客户端
        client = mqtt.Client()
        client.on_connect = on_connect
        client.on_disconnect = on_disconnect
        client.on_message = on_message
        client.on_subscribe = on_subscribe

        # 解析 broker_url 并连接
        # 示例 URL 格式: mqtt://username:password@host:port
        from urllib.parse import urlparse
        url = urlparse(broker_url)
        if (url.username):
            client.username_pw_set(url.username, url.password)
            logger.info(f"设置MQTT认证信息 - ID: {subscription_id}, Username: {url.username}")

        # 添加连接状态标志
        connected = threading.Event()

        # 设置自动重连。当客户端与服务器断开连接时，它不会立即尝试重新连接，而是根据设定的最小延迟时间开始等待。
        # 每次重连失败后，延迟时间会逐渐增加，直到达到最大延迟时间。
        # 这种机制可以避免在网络不稳定或服务器暂时不可用的情况下频繁尝试连接，从而减少不必要的资源消耗和网络压力。
        client.reconnect_delay_set(min_delay=1, max_delay=60)

        try:
            logger.info(f"开始连接MQTT服务器 - ID: {subscription_id}, Host: {url.hostname}, Port: {url.port or 1883}")
            client.connect(url.hostname, url.port or 1883)
            client.loop_start()

            # 等待连接成功
            if not connected.wait(timeout=10):
                logger.error(f"MQTT 连接超时 - ID: {subscription_id}")
                return

            # 使用事件来控制任务的运行
            stop_event = threading.Event()

            def check_subscription_status():
                while not stop_event.is_set():
                    try:
                        subscription.refresh_from_db()
                        if not subscription.is_active:
                            logger.info(f"MQTT 订阅已停用 - ID: {subscription_id}")
                            stop_event.set()
                            break
                        # 检查连接状态
                        if not connected.is_set():
                            logger.warning(f"MQTT 连接已断开，等待重连 - ID: {subscription_id}")
                        threading.Event().wait(5)  # 使用事件等待而不是 sleep
                    except Exception as e:
                        logger.error(f"检查订阅状态时出错 - ID: {subscription_id}, 错误: {str(e)}", exc_info=True)
                        break

            # 启动状态检查线程
            status_thread = threading.Thread(target=check_subscription_status)
            status_thread.daemon = True
            status_thread.start()
            logger.info(f"MQTT 订阅状态检查线程已启动 - ID: {subscription_id}")

            # 等待停止信号
            stop_event.wait()

        except Exception as e:
            logger.error(f"MQTT 连接过程出错 - ID: {subscription_id}, 错误: {str(e)}", exc_info=True)
        finally:
            # 清理资源
            logger.info(f"开始清理MQTT资源 - ID: {subscription_id}")
            client.loop_stop()
            client.disconnect()
            logger.info(f"MQTT 订阅任务已完成 - ID: {subscription_id}")

    except Exception as e:
        logger.error(f"MQTT 订阅任务失败 - ID: {subscription_id}, 错误: {str(e)}", exc_info=True)
